package mq

import (
	"context"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/segmentio/kafka-go"
)

// KafkaPlugin 实现基于 Kafka 的消息队列
type KafkaPlugin struct {
	brokers         []string
	closed          bool
	startTime       int64
	errorCount      int
	lastError       string
	queueStats      map[string]*QueueStats
	statsMutex      sync.RWMutex
	consumersMutex  sync.Mutex
	activeConsumers map[string]*kafka.Reader
}

// 确保 KafkaPlugin 实现了 MQPlugin 接口
var _ MQPlugin = (*KafkaPlugin)(nil)

// newKafkaPlugin 创建一个新的 KafkaPlugin 实例
func newKafkaPlugin(config map[string]interface{}) (MQPlugin, error) {
	// 从配置中获取 Kafka 服务器地址
	brokersStr, ok := config["brokers"].(string)
	if !ok || brokersStr == "" {
		return nil, fmt.Errorf("Kafka brokers 配置缺失或无效")
	}

	// 分割多个 broker 地址
	brokers := strings.Split(brokersStr, ",")
	for i, broker := range brokers {
		brokers[i] = strings.TrimSpace(broker)
	}

	return &KafkaPlugin{
		brokers:         brokers,
		closed:          false,
		startTime:       time.Now().Unix(),
		queueStats:      make(map[string]*QueueStats),
		activeConsumers: make(map[string]*kafka.Reader),
	}, nil
}

// Type 返回消息队列类型
func (p *KafkaPlugin) Type() MQType {
	return Kafka
}

// Publish 发布消息到指定主题
func (p *KafkaPlugin) Publish(ctx context.Context, topic string, message string) error {
	w := &kafka.Writer{
		Addr:     kafka.TCP(p.brokers...),
		Topic:    topic,
		Balancer: &kafka.Hash{},
	}

	defer w.Close()

	msg := kafka.Message{
		Value: []byte(message),
	}

	if err := w.WriteMessages(ctx, msg); err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("向 Kafka 发布消息失败: %v", err)
		log.Printf(p.lastError)
		return fmt.Errorf("向 Kafka 发布消息失败: %w", err)
	}

	return nil
}

// Subscribe 订阅指定主题的消息
func (p *KafkaPlugin) Subscribe(ctx context.Context, topic string, handler func(message string) error) error {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  p.brokers,
		Topic:    topic,
		GroupID:  "task-group",
		MinBytes: 1e3, // 1KB
		MaxBytes: 1e6, // 1MB
	})

	// 保存消费者到活跃消费者映射
	p.consumersMutex.Lock()
	p.activeConsumers[topic] = r
	p.consumersMutex.Unlock()

	// 初始化或更新队列统计信息
	p.statsMutex.Lock()
	if _, exists := p.queueStats[topic]; !exists {
		p.queueStats[topic] = &QueueStats{
			TopicName: topic,
		}
	}
	p.statsMutex.Unlock()

	go func() {
		defer r.Close()

		for {
			select {
			case <-ctx.Done():
				return
			default:
				msg, err := r.ReadMessage(ctx)
				if err != nil {
					p.errorCount++
					p.lastError = fmt.Sprintf("从 Kafka 读取消息失败: %v", err)
					log.Printf(p.lastError)
					continue
				}

				// 更新统计信息
				p.statsMutex.Lock()
				stats := p.queueStats[topic]
				stats.LastConsumedAt = time.Now().Unix()
				stats.ConsumedMessages++
				p.statsMutex.Unlock()

				if err := handler(string(msg.Value)); err != nil {
					log.Printf("处理消息失败: %v", err)

					// 更新失败统计
					p.statsMutex.Lock()
					stats := p.queueStats[topic]
					stats.FailedMessages++
					p.statsMutex.Unlock()
				}
			}
		}
	}()

	return nil
}

// Close 关闭 Kafka 连接
func (p *KafkaPlugin) Close() error {
	p.closed = true

	// 关闭所有活跃的消费者
	p.consumersMutex.Lock()
	for _, consumer := range p.activeConsumers {
		consumer.Close()
	}
	p.activeConsumers = make(map[string]*kafka.Reader)
	p.consumersMutex.Unlock()

	return nil
}

// GetStatus 获取Kafka的状态
func (p *KafkaPlugin) GetStatus() MQStatus {
	// 尝试建立连接以检查连接状态
	isConnected := false

	// 创建一个临时的连接来检查状态
	dialer := &kafka.Dialer{
		Timeout: 5 * time.Second,
	}

	for _, broker := range p.brokers {
		conn, err := dialer.DialContext(context.Background(), "tcp", broker)
		if err == nil {
			conn.Close()
			isConnected = true
			break
		}
	}

	if !isConnected {
		p.errorCount++
		p.lastError = "无法连接到任何 Kafka broker"
	}

	return MQStatus{
		IsConnected:    isConnected,
		ConnectionInfo: strings.Join(p.brokers, ","),
		StartTime:      p.startTime,
		ErrorCount:     p.errorCount,
		LastError:      p.lastError,
	}
}

// GetQueueStats 获取指定主题的队列统计信息
func (p *KafkaPlugin) GetQueueStats(topic string) (QueueStats, error) {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	// 检查队列是否存在
	stats, exists := p.queueStats[topic]
	if !exists {
		return QueueStats{}, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}

	// 尝试从Kafka获取实时队列信息
	// 注意：Kafka不像RabbitMQ那样容易获取队列长度，这里可能需要使用Admin API
	// 这里简化处理，只返回已有的统计信息

	return *stats, nil
}

// ListQueueStats 获取所有主题的队列统计信息
func (p *KafkaPlugin) ListQueueStats() []QueueStats {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	result := make([]QueueStats, 0, len(p.queueStats))
	for _, stats := range p.queueStats {
		result = append(result, *stats)
	}

	return result
}

// GetPendingMessages 获取待处理的消息
func (p *KafkaPlugin) GetPendingMessages(topic string, page, pageSize int) ([]string, error) {
	// 实现获取待处理消息的逻辑
	return []string{}, nil
}

// GetConsumedMessages 获取已消费的消息
func (p *KafkaPlugin) GetConsumedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 实现获取已消费消息的逻辑
	return []MessageRecord{}, nil
}

// GetFailedMessages 获取失败的消息
func (p *KafkaPlugin) GetFailedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 实现获取失败消息的逻辑
	return []MessageRecord{}, nil
}

// CloseQueue 关闭指定主题的队列
func (p *KafkaPlugin) CloseQueue(topic string) error {
	// 检查连接状态
	if p.closed {
		return fmt.Errorf("Kafka 连接已关闭")
	}

	// 从活跃消费者中查找并关闭
	p.consumersMutex.Lock()
	defer p.consumersMutex.Unlock()

	consumer, exists := p.activeConsumers[topic]
	if !exists {
		// 如果没有活跃的消费者，只从统计信息中移除
		p.statsMutex.Lock()
		delete(p.queueStats, topic)
		p.statsMutex.Unlock()
		return nil
	}

	// 关闭消费者
	err := consumer.Close()
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("关闭 Kafka 消费者失败: %v", err)
		return fmt.Errorf("关闭 Kafka 消费者失败: %w", err)
	}

	// 从活跃消费者映射中移除
	delete(p.activeConsumers, topic)

	// 从统计信息中移除
	p.statsMutex.Lock()
	delete(p.queueStats, topic)
	p.statsMutex.Unlock()

	return nil
}

// ListTopics 获取所有主题列表
func (p *KafkaPlugin) ListTopics() []string {
	// 尝试从Kafka服务器获取所有主题
	conn, err := kafka.Dial("tcp", p.brokers[0])
	if err == nil {
		defer conn.Close()

		partitions, err := conn.ReadPartitions()
		if err == nil {
			// 使用map去重
			topicMap := make(map[string]struct{})
			for _, p := range partitions {
				// 排除内部主题
				if !strings.HasPrefix(p.Topic, "__") {
					topicMap[p.Topic] = struct{}{}
				}
			}

			// 转换为切片
			topics := make([]string, 0, len(topicMap))
			for topic := range topicMap {
				topics = append(topics, topic)
			}
			return topics
		}
	}

	// 如果无法从服务器获取，则返回已知的主题
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	topics := make([]string, 0, len(p.queueStats))
	for topic := range p.queueStats {
		topics = append(topics, topic)
	}

	return topics
}
